package com.zhengb.flink.study.datastream.datasource;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.concurrent.ThreadLocalRandom;

/**
 * rich source 生成随机数
 *
 * @author zhengbo
 * @date 2019/12/19
 */
@Slf4j
public class RandomParallelSource extends RichParallelSourceFunction<Long> {
    private static volatile boolean isRunning = true;

    @Override
    public void run(SourceContext<Long> ctx) throws Exception {

        while (isRunning) {
            long randomNumber = ThreadLocalRandom.current().nextLong(1000);

            log.info("generate random number:{}", randomNumber);

            ctx.collectWithTimestamp(randomNumber, System.currentTimeMillis());

            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
